/*******************************************************************************
 * @(#) HBaseUtils.java  2017/9/11
 *
 * Copyright 2016 emrubik Group Ltd. All rights reserved.
 * EMRubik PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 *******************************************************************************/

package com.example.hbase.common;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;

/**
 * @author <a href="mailto:wangjh@emrubik.com">王江欢</a>
 * @version $Revision 1.0 $ 2017/9/11
 */
public class HBaseUtils {

    private static Configuration conf = HBaseConfiguration.create();

    private static ExecutorService poolx = Executors.newFixedThreadPool(300);

    static {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.zookeeper.quorum", "cdh0");
    }

    public static Connection getConnection() {
        int i = 0;
        Connection conn = null;
        do {
            try {
                conn = ConnectionFactory.createConnection(conf, poolx);
                if (conn != null) {
                    break;
                }
                Thread.sleep(100);
                i++;
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (IOException e) {
                e.printStackTrace();
            }
        } while (conn == null && i < 5);
        return conn;
    }

    public static void closeConnection(Connection connection) {
        try {
            connection.close();
            poolx.shutdownNow();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void main(String args[]) {
        HBaseUtils hu = new HBaseUtils();
        hu.run();
    }

    private void run(){
        try {
            Thread[] threads = new Thread[10];
            for (int i = 0; i < threads.length; i++) {
                threads[i] = new ImportThread();
                threads[i].start();
            }
            for (int j = 0; j < threads.length; j++) {
                (threads[j]).join();
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public class ImportThread extends Thread {
        public void run() {
            System.out.println(HBaseConnectionPool.getInstance(10,30).getConnection().getConnection().hashCode());
        }
    }
}
